package com.demo;

import com.demo.config.SyncConfig;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.MemorySize;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.configuration.CoreOptions;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class FlinkMySQLToGaussSync {
    private static final SyncConfig config = SyncConfig.getInstance();
    private static final ObjectMapper objectMapper = new ObjectMapper();

    public static void main(String[] args) throws Exception {
        log.info("开始初始化 MySQL 到 GaussDB 的同步任务...");
        
        // 1. 创建Flink执行环境（带配置）
        Configuration configuration = new Configuration();
        // 进一步优化资源配置
        configuration.set(TaskManagerOptions.TASK_HEAP_MEMORY, MemorySize.parse("128m"));
        configuration.set(TaskManagerOptions.MANAGED_MEMORY_SIZE, MemorySize.parse("64m"));
        configuration.setString("taskmanager.memory.network.min", "16mb");
        configuration.setString("taskmanager.memory.network.max", "16mb");
        configuration.setString("taskmanager.memory.framework.off-heap.size", "32mb");
        configuration.setInteger("taskmanager.numberOfTaskSlots", 1);
        
        // 设置最小/最大并行度
        configuration.setInteger(CoreOptions.DEFAULT_PARALLELISM.key(), 1);
        configuration.setInteger("parallelism.max", 1);
        
        // 禁用JVM指标跟踪以减少内存使用
        configuration.setString("metrics.jvm.gc.disabled", "true");
        configuration.setString("metrics.jvm.memory.disabled", "true");
        configuration.setString("metrics.jvm.threads.disabled", "true");
        
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);
        
        // 设置全局并行度为1
        env.setParallelism(1);
        env.setMaxParallelism(1);
        
        // 2. 配置checkpoint
        configureCheckpoint(env);
        
        // 3. 配置MySQL CDC Source
        MySqlSource<String> mySqlSource = configureMySqlSource();
        
        // 4. 创建数据流并处理
        createAndProcessDataStream(env, mySqlSource);
        
        // 5. 启动执行
        env.execute("MySQL to GaussDB Sync Job");
    }

    private static void configureCheckpoint(StreamExecutionEnvironment env) {
        // 配置checkpoint
        env.enableCheckpointing(config.getCheckpointInterval());
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(1000);
        env.getCheckpointConfig().setCheckpointTimeout(60000);
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
        env.getCheckpointConfig().setTolerableCheckpointFailureNumber(3);
        env.getCheckpointConfig().setExternalizedCheckpointCleanup(
            CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        
        // 配置重启策略 - 增加重试次数和间隔
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(5, 20000));
        
        log.info("Checkpoint配置完成: 间隔={}ms, 模式=EXACTLY_ONCE", config.getCheckpointInterval());
    }

    private static MySqlSource<String> configureMySqlSource() {
        String sourceTable = config.getMySQLDatabase() + "." + config.getMySQLTable();
        log.info("配置 MySQL Source: {}", sourceTable);
        
        return MySqlSource.<String>builder()
                .hostname(config.getMySQLHost())
                .port(config.getMySQLPort())
                .databaseList(config.getMySQLDatabase())
                .tableList(sourceTable)
                .username(config.getMySQLUsername())
                .password(config.getMySQLPassword())
                .deserializer(new JsonDebeziumDeserializationSchema())
                .startupOptions(StartupOptions.initial()) // 全量+增量
                .serverTimeZone("Asia/Shanghai")
                .build();
    }

    private static void createAndProcessDataStream(
            StreamExecutionEnvironment env, 
            MySqlSource<String> mySqlSource) {
        
        log.info("创建数据流处理管道...");
        
        // 创建数据源
        DataStreamSource<String> stream = env.fromSource(
                mySqlSource,
                WatermarkStrategy.noWatermarks(),
                "MySQL CDC Source"
        );

        // 转换并写入GaussDB
        stream.map(new RecordTransformer())
              .name("Record Transformer")
              .addSink(createGaussSink())
              .name("GaussDB Sink");
        
        log.info("数据流处理管道创建完成");
    }

    private static class RecordTransformer implements MapFunction<String, String> {
        @Override
        public String map(String value) throws Exception {
            try {
                JsonNode jsonNode = objectMapper.readTree(value);
                JsonNode after = jsonNode.get("after");
                
                if (after != null && after.has("name")) {
                    String name = after.get("name").asText();
                    String upperName = name.toUpperCase();
                    ((ObjectNode) after).put("name", upperName);
                    log.info("转换name: {} -> {}", name, upperName);
                }
                
                return objectMapper.writeValueAsString(jsonNode);
            } catch (Exception e) {
                log.error("数据转换失败: {}", value, e);
                throw e;
            }
        }
    }

    private static SinkFunction<String> createGaussSink() {
        log.info("配置 GaussDB Sink: {}", config.getGaussURL());
        
        // 使用MERGE INTO语法
        String insertSql = String.format(
            "MERGE INTO %s t1 " +
            "USING (VALUES (?, ?, ?)) AS t2(id, name, gender) " +
            "ON t1.id = t2.id " +
            "WHEN MATCHED THEN " +
            "  UPDATE SET name = t2.name, gender = t2.gender " +
            "WHEN NOT MATCHED THEN " +
            "  INSERT (id, name, gender) VALUES (t2.id, t2.name, t2.gender)",
            config.getGaussTable()
        );

        return JdbcSink.sink(
                insertSql,
                (statement, record) -> {
                    try {
                        JsonNode jsonNode = objectMapper.readTree(record);
                        JsonNode after = jsonNode.get("after");
                        
                        if (after == null) {
                            log.warn("跳过无after数据的记录: {}", record);
                            return;
                        }
                        
                        statement.setInt(1, after.get("id").asInt());
                        statement.setString(2, after.get("name").asText());
                        statement.setString(3, after.get("gender").asText());
                        
                        log.debug("准备写入数据: id={}, name={}, gender={}",
                                after.get("id").asInt(),
                                after.get("name").asText(),
                                after.get("gender").asText());
                    } catch (Exception e) {
                        log.error("处理记录失败: {}", record, e);
                        throw new RuntimeException("处理记录失败", e);
                    }
                },
                JdbcExecutionOptions.builder()
                        .withBatchSize(config.getBatchSize())
                        .withBatchIntervalMs(config.getBatchInterval())
                        .withMaxRetries(config.getMaxRetries())
                        .build(),
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withUrl(config.getGaussURL())
                        .withDriverName("org.opengauss.Driver")
                        .withUsername(config.getGaussUsername())
                        .withPassword(config.getGaussPassword())
                        .build()
        );
    }
} 